package org.future.task.redis.connection;

import org.future.task.core.handler.Commit;
import org.future.task.core.handler.Connection;
import org.future.task.core.serialize.MineSerializer;
import org.future.task.redis.factory.RedisTemplateFactory;
import redis.clients.jedis.JedisPoolConfig;

import java.util.List;
import java.util.Random;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author Dongwei-Chen
 * @Date 2023/1/9 10:31
 * @Description
 */
public class MutiRedisConnectionImpl implements Connection {

    private AtomicInteger atom = new AtomicInteger(0);

    private MineSerializer mineSerializer;

    private List<String> hostAndPorts;

    private JedisPoolConfig jedisPoolConfig;

    private int commitThreadSize;

    private CopyOnWriteArrayList<Connection> connections;

    public MutiRedisConnectionImpl(List<String> hostAndPorts, JedisPoolConfig jedisPoolConfig, int writeBatch, int localCacheSize) throws IllegalAccessException, InstantiationException, ClassNotFoundException {
        this(hostAndPorts, jedisPoolConfig, null, writeBatch, localCacheSize, 0);
    }

    public MutiRedisConnectionImpl(List<String> hostAndPorts, JedisPoolConfig jedisPoolConfig, MineSerializer mineSerializer, int writeBatch) throws IllegalAccessException, InstantiationException, ClassNotFoundException {
        this(hostAndPorts, jedisPoolConfig, mineSerializer, writeBatch, 0, 0);
    }

    public MutiRedisConnectionImpl(List<String> hostAndPorts, JedisPoolConfig jedisPoolConfig, int writeBatch, int localCacheSize, int commitThreadSize) throws IllegalAccessException, InstantiationException, ClassNotFoundException {
        this(hostAndPorts, jedisPoolConfig, null, writeBatch, localCacheSize, commitThreadSize);
    }

    public MutiRedisConnectionImpl(List<String> hostAndPorts, JedisPoolConfig jedisPoolConfig, MineSerializer mineSerializer, int writeBatch, int commitThreadSize) throws IllegalAccessException, InstantiationException, ClassNotFoundException {
        this(hostAndPorts, jedisPoolConfig, mineSerializer, writeBatch, 0, commitThreadSize);
    }

    public MutiRedisConnectionImpl(List<String> hostAndPorts, JedisPoolConfig jedisPoolConfig, MineSerializer mineSerializer, int writeBatch, int localCacheSize, int commitThreadSize) throws IllegalAccessException, InstantiationException, ClassNotFoundException {
        this.mineSerializer = mineSerializer;
        this.hostAndPorts = hostAndPorts;
        this.jedisPoolConfig = jedisPoolConfig;
        connections = RedisTemplateFactory.getRedisTemplate(hostAndPorts, jedisPoolConfig, mineSerializer, writeBatch, localCacheSize,commitThreadSize);
    }

    public MutiRedisConnectionImpl(List<String> hostAndPorts, JedisPoolConfig jedisPoolConfig, MineSerializer mineSerializer) throws IllegalAccessException, InstantiationException, ClassNotFoundException {
        this(hostAndPorts, jedisPoolConfig, mineSerializer, 0);
    }

    public MutiRedisConnectionImpl(List<String> hostAndPorts) throws IllegalAccessException, ClassNotFoundException, InstantiationException {
        this(hostAndPorts, null, null);
    }

    public MutiRedisConnectionImpl(List<String> hostAndPorts, JedisPoolConfig jedisPoolConfig) throws IllegalAccessException, ClassNotFoundException, InstantiationException {
        this(hostAndPorts, jedisPoolConfig, null);
    }

    public MutiRedisConnectionImpl(List<String> hostAndPorts, JedisPoolConfig jedisPoolConfig, int writeBatch) throws IllegalAccessException, ClassNotFoundException, InstantiationException {
        this(hostAndPorts, jedisPoolConfig, null, writeBatch);
    }

    public MutiRedisConnectionImpl(List<String> hostAndPorts, MineSerializer mineSerializer) throws IllegalAccessException, ClassNotFoundException, InstantiationException {
        this(hostAndPorts, null, mineSerializer);
    }

    @Override
    public List<Object> getDataList(String queueName, int size, int timeOut) {
        Connection connection = connections.get(atom.getAndIncrement() % connections.size());
        List<Object> dataList = connection.getDataList(queueName, size, 100);

        return dataList;
    }

    @Override
    public long size(String queueName) {
        long size = 0;
        for (Connection connection : connections) {
            size += connection.size(queueName);
        }
        return size;
    }

    @Override
    public void sendDataList(String queueName, List<?> dataList) throws Exception {
        int next = new Random().nextInt(connections.size());
        Connection connection = connections.get(next);
        connection.sendDataList(queueName, dataList);
    }

    @Override
    public void sendData(String queueName, Object data) throws InterruptedException {
        int next = new Random().nextInt(connections.size());
        Connection connection = connections.get(next);
        connection.sendData(queueName, data);
    }

    @Override
    public String getType() {
        return connections.getClass().getSimpleName();
    }

    @Override
    public Commit getCommit() {
        return null;
    }

}
